Tracking issue

N/A

Why are the changes needed?

Currently, timeouts are not supported on ArryaNode subNodes. If there is a timeout set in the task decorator, that is applied only to the overall ArrayNode and not individual subNodes. This change ensures subNodes are executed with timeouts as a standalone execution would be.

What changes were proposed in this pull request?

We store a deltaTimestamp to track startedAt for subNode executions. This delta is computed from the overall ArrayNode start time. This is stored using a bitarray (just like other ArrayNode subNode metadata), with a preset 18 bits of precision. This means for 5k subNodes there will be approximately 90kb added to the FlyteWorkflow CR - reducing scalability by a small amount. The 18 bits gives us around 3 days of delta time, meaning subNodes start times can be tracked up to 3 days after the overall ArrayNode starts. If this proves to be too imprecise, we can increase this value and incur further reductions to scale.

It is important to note that we do not have a task phase for TIMED_OUT. So in the UI this will display as FAILED. This is a restriction of the current eventing scheme for map tasks.

How was this patch tested?

This was tested locally under a variety of failure scenarios (ex. retryable failures).

Setup process

from datetime import timedelta
from flytekit import map_task, task, workflow
from typing import List

import time
from flytekit.exceptions.user import FlyteRecoverableException


@task(retries=3, timeout=timedelta(minutes=1))
def say_hello(name: str) -> str:
    #time.sleep(120)

    time.sleep(10)
    raise FlyteRecoverableException(f"testing failure for {name}")

    return f"hello {name}"


@workflow
def say_hello_wf(name: List[str]) -> List[str]:
    return map_task(say_hello)(name=name)

Screenshots

N/A

Check all the applicable boxes

• I updated the documentation accordingly.
• All new and existing tests passed.
• All commits are signed-off.

Related PRs

N/A

Docs link

N/A